package com.atguigu.gmall.realtime.dim.app;

/*import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.HBaseSinkFunction;
import com.atguigu.gmall.realtime.common.function.HbaseSinkFunction;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import io.debezium.data.Json;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.function.Predicate;

*//**
 * 对维度数据的处理
 * 总结:
 * 1. DIM处理流程
 * 模拟生成数据 -> Mysql -> binlog -> MaxWell -> topic_db  ->  KafkaSource -> 清洗过滤、转换结构
 * -> connect -> 过滤维度数据（广播状态 、 预加载配置表）  -> 写入HBase
 * FlinkCDC   ->  HBase建表删表
 * 2. 重点
 * 1) KafkaSource的使用
 * 2) FlinkCDC的使用  以及 配置表的思想  以及 解决了什么问题
 * 3) 广播状态的使用  以及 解决了什么问题
 * 4) 基本API的使用 ， Flink的算子 、 HBase的API 、 JDBC的基本操作等。
 * <p>
 * 对维度数据的处理
 * <p>
 * 在open方法中预加载配置表数据,将读取到的数据维护到一个普通的map集合中
 * 在处理数据流汇总的数据的时候, 优先从状态中获取配置信息, 如果获取不到, 再尝试从map中获取
 * <p>
 * 主要解决程序刚启动, 配置表的数据还没有被读取过来, 状态中还没有维护好配置信息, 数据流中的数据已经过来了
 * 没有正确的将维度表的数据过滤出来
 * <p>
 * 对数据流中的数据进行处理
 * 判断当前数据是否是来自于维度表的数据 , 如果是维度表的数据, 写出,反之不要了
 * <p>
 * 对广播流 (配置流) 中的数据进行处理
 * 将广播流中过来的数据维护到广播状态中
 * <p>
 * r c u的操作: 更新(添加或覆盖)状态中的额数据
 * d: 删除状态中对应的数据
 * <p>
 * <p>
 * 在open方法中，获取Hbase的Connection， 保证每个并行实例中只获取一次Connection对象
 * <p>
 * 在close方法中，关闭Hbase的Connection
 *//*

 *//**
 * 对维度数据的处理
 *//*
@Slf4j
public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().start(10011, 4, "dim_app", Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {


        // 对数据的处理

        // 对数据进行清洗过滤, 并转换成JSONObject结构
        // 我们会用到的数据: Maxwell全量同步数据(bootstrap_insert),增删改insert,update,delete
        SingleOutputStreamOperator<JSONObject> etlStream = elt(stream);
        // etlStream.print("INPUT");

        // 基于FlinkCDC读取配置表数据
        SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs = readTabelProcessDim(env);
        // tableProcessDimDs.print();


        // 到HBase中进行建表,  删表
        SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs = createOrDropTable(tableProcessDimDs);
        // createOrDropTableDs.print();


        //  过滤出维度数据
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDS = filterDimData(etlStream, createOrDropTableDs);
        // filterDimDataDS.print("DIM");


        // 将维度数据写出到HBase中
        // 需要自定义SinkFunction, 来完成写入HBase的操作
        writeToHBase(filterDimDataDS);

    }

    private static void writeToHBase(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDS) {
        filterDimDataDS.addSink(
                *//*new RichSinkFunction<Tuple2<JSONObject, TableProcessDim>>() {
                    private Connection connection;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connection = HBaseUtil.getConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeConnection(connection);
                    }

                    @Override
                    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
                        // 提取待写入数据
                        JSONObject dataJsonObj = value.f0;
                        // 配置对象
                        TableProcessDim tableProcessDim = value.f1;

                        // 提取并移除type
                        String type = dataJsonObj.remove("type").toString();
                        // 提取SinkTable
                        String sinkTable = tableProcessDim.getSinkTable();
                        // 提取SinkColumns
                        String sinkColumns = tableProcessDim.getSinkColumns();
                        // 提取SinkFamily
                        String sinkFamily = tableProcessDim.getSinkFamily();
                        // 提取SinkRowKey
                        // String sinkRowKey = tableProcessDim.getSinkRowKey(); 错误,不能这样提取,要的不是名字,要的是那个值
                        // 先提取RowKey列,再提取对应的值
                        String sinkRowKeyColumn = tableProcessDim.getSinkRowKey();
                        String rowKey = dataJsonObj.getString(sinkRowKeyColumn);

                        if (type.equals(Constant.MAXWELL_TYPE_DELETE)){
                            // type : delete
                            // 执行删除维度数据的操作
                            HBaseUtil.deleteRow(connection, Constant.HBASE_NAMESPACE, sinkTable, rowKey);

                        }else {
                            // type : insert , update, bootstrap_insert
                            // 执行写入维度数据的操作
                            HBaseUtil.putRow(connection, Constant.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, dataJsonObj);
                        }

                    }
                }*//*
                new HBaseSinkFunction()
        );
    }

    private static SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimData(SingleOutputStreamOperator<JSONObject> etlStream, SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs) {
        // 对维度数据进行处理
        // 将配置表的流进行广播处理, 然后 数据流 再与 广播留进行connect处理, 分别对两条流的数据进行处理,
        // 将广播流中的数据维护到广播状态中, 处理数据流的 过程中. 可以基于广播状态中维护的配置表信息来判断当前是否为来自维度表的数据

        // 将配置流处理成广播流
        // key: source_table value: TableProcessDim
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor
                = new MapStateDescriptor<>("mapState", Types.STRING, Types.POJO(TableProcessDim.class));
        BroadcastStream<TableProcessDim> broadcastStream = createOrDropTableDs.broadcast(mapStateDescriptor);

        // connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDS = etlStream.connect(broadcastStream)
                .process(
                        new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {
                            *//**
 * 在open方法中预加载配置表数据,将读取到的数据维护到一个普通的map集合中
 * 在处理数据流汇总的数据的时候, 优先从状态中获取配置信息, 如果获取不到, 再尝试从map中获取
 *
 * 主要解决程序刚启动, 配置表的数据还没有被读取过来, 状态中还没有维护好配置信息, 数据流中的数据已经过来了
 * 没有正确的将维度表的数据过滤出来
 *//*
                            private Map<String, TableProcessDim> configMap = new HashMap<>();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                // 预加载配置表的数据
                                // 通过jdbc的方式实现
                                java.sql.Connection connection = JdbcUtil.getConnection();
                                List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(connection,
                                        "select source_table, sink_table, sink_family, sink_columns, sink_row_key " +
                                                "from "+Constant.GMALL_CONFIG_DB+"."+Constant.TABLE_PROCESS_DIM,
                                        TableProcessDim.class, true);

                                *//*ArrayList<TableProcessDim> tableProcessDimList = new ArrayList<>();
                                // 1.注册启动
                                Class.forName("com.mysql.cj.jdbc.Driver");
                                // 2.获取连接
                                java.sql.Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall2024_config_240228", "root", "000000");
                                // 3. 编写sql
                                String sql = "select source_table, sink_table, sink_family, sink_columns, sink_row_key from gmall2024_config_240228.table_process_dim ";
                                // 4. 编译sql
                                PreparedStatement preparedStatement = connection.prepareStatement(sql);
                                // 5. 执行sql
                                ResultSet resultSet = preparedStatement.executeQuery();
                                // 6. 处理结果集
                                while (resultSet.next()) {
                                    TableProcessDim tableProcessDim = new TableProcessDim();
                                    tableProcessDim.setSourceTable(resultSet.getString("source_table"));
                                    tableProcessDim.setSinkTable(resultSet.getString("sink_table"));
                                    tableProcessDim.setSinkFamily(resultSet.getString("sink_family"));
                                    tableProcessDim.setSinkColumns(resultSet.getString("sink_columns"));
                                    tableProcessDim.setSinkRowKey(resultSet.getString("sink_row_key"));

                                    tableProcessDimList.add(tableProcessDim);
                                }
                                // 7. 返回结果*//*

                                // 将数据维护到Map中
                                tableProcessDimList.forEach(tableProcessDim -> configMap.put(tableProcessDim.getSourceTable(), tableProcessDim));

                            }

                            *//**
 * 对数据流中的数据进行处理
 * 判断当前数据是否是来自于维度表的数据 , 如果是维度表的数据, 写出,反之不要了
 *//*
                            @Override
                            public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
                                // 获取广播状态
                                ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);


                                // 从数据中提取表名
                                String tableName = jsonObject.getString("table");

                                // 判断是否为维度表数据
                                TableProcessDim tableProcessDim = null;
                                if ((tableProcessDim = broadcastState.get(tableName)) != null // 是维度表数据
                                    || ((tableProcessDim = configMap.get(tableName)) != null)) {

                                    // 只需要将data中的数据写出, 提取data
                                    JSONObject dataJsonObj = jsonObject.getJSONObject("data");

                                    // 删除不需要的字段
                                    // 先从tableProcessDim同去需要的字段
                                    List<String> needColumns = Arrays.asList(tableProcessDim.getSinkColumns().split(","));
                                    dataJsonObj.keySet().removeIf(
                                            new Predicate<String>() {
                                                @Override
                                                public boolean test(String key) {
                                                    return needColumns.contains(key);
                                                }
                                            }
                                    );


                                    // 补充type, 下游将数据写入到HBase时, 需要基于type决定做put还是delete
                                    dataJsonObj.put("type", jsonObject.getString("type"));


                                    // 写出数据
                                    collector.collect(Tuple2.of(dataJsonObj, tableProcessDim));

                                }

                            }

                            *//**
 * 对广播流 (配置流) 中的数据进行处理
 * 将广播流中过来的数据维护到广播状态中
 *
 * r c u的操作: 更新(添加或覆盖)状态中的额数据
 * d: 删除状态中对应的数据
 *//*
                            @Override
                            public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
                                // 获取广播状态
                                BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);

                                // 状态中的key
                                String key = tableProcessDim.getSourceTable();
                                // 提取操作类型
                                String op = tableProcessDim.getOp();

                                if (op.equals(Constant.CDC_OP_D)) {
                                    // d 的操作
                                    // 从状态中当前key对应的数据清除掉
                                    broadcastState.remove(key);

                                    // 从configMap中删除
                                    configMap.remove(key);
                                } else {
                                    // r c u 的操作
                                    // 更新状态中的key的数据
                                    broadcastState.put(key, tableProcessDim);
                                    // 更新configMap中key的数据
                                    configMap.put(key, tableProcessDim);
                                }

                            }
                        }
                );
        return filterDimDataDS;
    }

    private static SingleOutputStreamOperator<TableProcessDim> createOrDropTable(SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs) {
        SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs = tableProcessDimDs.process(
                new ProcessFunction<TableProcessDim, TableProcessDim>() {
                    private Connection connection;

                    *//**
 *
 * 在open方法中，获取Hbase的Connection， 保证每个并行实例中只获取一次Connection对象
 *//*
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connection = HBaseUtil.getConnection();
                    }

                    *//**
 * 在close方法中，关闭Hbase的Connection
 *//*
                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeConnection(connection);
                    }

                    @Override
                    public void processElement(TableProcessDim tableProcessDim, ProcessFunction<TableProcessDim, TableProcessDim>.Context context, Collector<TableProcessDim> collector) throws Exception {
                        // 根据TableProcessDim中的op决定是建表还是删表
                        // op:r,c -> 建表
                        // op:u -> 先删表再建表
                        // op:d -> 删表

                        // 提取类型
                        String op = tableProcessDim.getOp();
                        // 提取SinkTable
                        String sinkTable = tableProcessDim.getSinkTable();
                        // 提取列族
                        String sinkFamily = tableProcessDim.getSinkFamily();
                        if (op.equals(Constant.CDC_OP_D)) {
                            // op:d -> 删表
                            HBaseUtil.dropTable(connection, Constant.HBASE_NAMESPACE, sinkTable);
                        } else if (Constant.CDC_OP_U.equals(op)) {
                            // op:u -> 先删表再建表
                            HBaseUtil.dropTable(connection, Constant.HBASE_NAMESPACE, sinkTable);
                            HBaseUtil.createTable(connection, Constant.HBASE_NAMESPACE, sinkTable, sinkFamily);
                        } else {
                            // op:r,c -> 建表
                            HBaseUtil.createTable(connection, Constant.HBASE_NAMESPACE, sinkTable, sinkFamily);
                        }
                        // 写出数据
                        collector.collect(tableProcessDim);
                    }
                }
        ).setParallelism(1);
        return createOrDropTableDs;
    }

    private static SingleOutputStreamOperator<TableProcessDim> readTabelProcessDim(StreamExecutionEnvironment env) {
        // 使用FLinkCDC 读取 维度配置表, 转换成 TableProcessDim类型
        // 如何维护维度表:
        // 方式一(x): 通过自定义集合, 维护所有的维度表  [...一些维度表],缺点:新增或者删除维度表不好对这个集合进行修改,除非是维度表变动不大
        // 方式二(x): 定义到配置文件中, 可以灵活修改配置文件, 需要解决如何让程序重新加载配置文件, 让配置中的内容生效
        // 方式三(x): 定义到数据库表中, 可以灵活修改表中的数据, 需要解决当数据库表中的数据发生改变后, 如何让程序重新读取新的数据
        // 实现方式:
        // 1. 每到达一条数据, 就到数据库读取一次最新的维度表信息 (读取太频繁,且大部分是重复数据 不推荐) (x)
        // 2. 如果数据库的数据发生变化, 需要读取, 不发生变化, 不用读取
        //      数据发生变化 -> binlog -> Maxwell -> Kafka -> KafkaSource -> stream (x)
        //      数据发送变化 -> binlog -> FlinkCDC -> stream (√)

        // 方式四: Zookeeper : 监听 + 通知 机制 ,目前不考虑此方法 (x)

        // 注意: 再mysql中创建gmall2024_config_240228的数据库,在数据库中创建一个table_process_dim表
        // 在 mysql中的 etc/my.cnf文件中添加binlog-do-db=gmall2024_config_240228配置, 重启mysql服务,sudo systemctl restart mysqld
        // 对mysql的etc/my.cnf文件进行修改会导致Maxwell出问题,需要将Maxwell在mysql数据库中的表删除掉, 重启Maxwell
        MySqlSource<String> mySqlSource = FlinkSourceUtil.getMySqlSource(Constant.GMALL_CONFIG_DB, Constant.TABLE_PROCESS_DIM);
        DataStreamSource<String> configDs = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource")
                .setParallelism(1);
        SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs = configDs.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String s) throws Exception {
                        // r:  数据在after中
                        // c:  数据在after中
                        // u:  数据在after中
                        // d:  数据在before中

                        // 将数据转换成Json格式
                        JSONObject jsonObject = JSON.parseObject(s);
                        // 提取操作类型
                        String op = jsonObject.getString("op");

                        TableProcessDim tableProcessDim = null;
                        if (Constant.CDC_OP_D.equals(op)) {
                            // 从before中提取数据 c r u
                            tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                        } else {
                            // 从after中提取数据 d
                            tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                        }

                        // 补充op
                        tableProcessDim.setOp(op);

                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        return tableProcessDimDs;
    }

    private static SingleOutputStreamOperator<JSONObject> elt(DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                        try {
                            // 将数据转化为Json格式
                            JSONObject jsonObject = JSON.parseObject(s);
                            // 提取database
                            String database = jsonObject.getString("database");
                            // 提取操作类型
                            String type = jsonObject.getString("type");
                            // 提取data
                            JSONObject dataJsonObj = jsonObject.getJSONObject("data");

                            // 判断类型是否符合
                            if (database.equals(Constant.DW_DB)
                                    && (
                                    Constant.MAXWELL_TYPE_INSERT.equals(type) ||
                                            Constant.MAXWELL_TYPE_UPDATE.equals(type) ||
                                            Constant.MAXWELL_TYPE_DELETE.equals(type) ||
                                            Constant.MAXWELL_TYPE_BOOTSTRAP_INSERT.equals(type)
                            )
                                    && dataJsonObj != null && dataJsonObj.size() >= 2
                            ) {
                                out.collect(jsonObject);
                            }


                        } catch (Exception e) {
                            System.out.println("DIM 脏数据" + s);
                            // log.warn("DIM 脏数据" + s);
                        }
                    }
                }
        );
        return etlStream;
    }
}*/

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.HBaseSinkFunction;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.util.*;

/**
 * @author WEIYUNHUI
 * @date 2024/7/31 11:44
 *
 * 对维度数据的处理
 *
 * 总结:
 *    1. DIM处理流程
 *
 *      模拟生成数据 -> Mysql -> binlog -> MaxWell -> topic_db  ->  KafkaSource -> 清洗过滤、转换结构
 *                                                                                             -> connect -> 过滤维度数据（广播状态 、 预加载配置表）  -> 写入HBase
 *                                                                FlinkCDC   ->  HBase建表删表
 *
 *    2. 重点
 *
 *       1) KafkaSource的使用
 *
 *       2) FlinkCDC的使用  以及 配置表的思想  以及 解决了什么问题
 *
 *       3) 广播状态的使用  以及 解决了什么问题
 *
 *       4) 基本API的使用 ， Flink的算子 、 HBase的API 、 JDBC的基本操作等。
 *
 *
 */
@Slf4j
public class DimApp extends BaseApp {

    public static void main(String[] args) {
        // 调用start()方法
        new DimApp().start(10001, 4, "dim_app1", Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //对数据的处理
        // 对数据进行清洗过滤，并转换成JSONObject结构
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);
        stream.print("kafka");
         etlStream.print("INPUT") ;

        //基于FlinkCDC读取配置表数据
        SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs = readTableProcessDim(env);
        tableProcessDimDs.print();


        //到HBase中进行建表 、 删表
        SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs = createOrDropTable(tableProcessDimDs);


        //过滤出维度数据
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDs = filterDimData(etlStream, createOrDropTableDs);
         filterDimDataDs.print("DIM");

        // 将维度数据写出到HBase中
        // 需要自定义SinkFunction， 来完成写入Hbase的操作
        writeToHBase(filterDimDataDs);
    }

    private static void writeToHBase(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDs) {
        filterDimDataDs.addSink(
                new HBaseSinkFunction()
        );
    }

    private static SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimData(SingleOutputStreamOperator<JSONObject> etlStream, SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs) {
        // 对维度数据进行处理
        // 将配置表的流进行广播处理， 然后 数据流 再与广播流进行connect ， 分别对两条流的数据进行处理，
        // 将广播流中的数据维护到广播状态中， 处理数据流的过程中， 可以基于广播状态中维护的配置表信息来判断当前是否为来自维度表的数据
        //将配置流处理成广播流
        // K: source_table  V : TableProcessDim
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor
                = new MapStateDescriptor<>("mapState", Types.STRING, Types.POJO(TableProcessDim.class));
        BroadcastStream<TableProcessDim> broadcastStream = createOrDropTableDs.broadcast(mapStateDescriptor);

        //connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterDimDataDs = etlStream.connect(broadcastStream)
                .process(
                        new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {

                            /**
                             * 在open方法中预加载配置表的数据， 将读取到的数据维护到一个普通的Map集合中
                             *
                             * 在处理数据流中的数据的时候， 优先从状态中获取配置信息，如果获取不到，再尝试从Map中获取。
                             *
                             * 主要解决程序刚启动， 配置表的数据还没有读取过来， 状态中还没有维护好配置信息， 数据流中的数据已经过来，
                             * 没有正确的将维度表的数据过滤出来。
                             */

                            private Map<String, TableProcessDim> configMap = new HashMap<>();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                // 预加载配置表的数据
                                // 通过JDBC的方式实现
                                java.sql.Connection connection = JdbcUtil.getConnection();

                                List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(
                                        connection,
                                        " SELECT " +
                                                " source_table , sink_table , sink_family , sink_columns , sink_row_key " +
                                                " FROM " + Constant.GMALL_CONFIG_DB + "." + Constant.TABLE_PROCESS_DIM,
                                        TableProcessDim.class,
                                        true
                                );

                                // 将数据维护到Map中
                                tableProcessDimList.forEach(tableProcessDim -> configMap.put(tableProcessDim.getSourceTable(), tableProcessDim));
                                JdbcUtil.closeConnection(connection);
                            }

                            /**
                             * 对数据流中的数据进行处理
                             *
                             * 判断当前数据是否是来自于维度表， 如果是维度表的数据， 写出
                             */
                            @Override
                            public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
                                //获取广播状态
                                ReadOnlyBroadcastState<String, TableProcessDim> broadcastState
                                        = ctx.getBroadcastState(mapStateDescriptor);

                                // 从数据中提取表名
                                String tableName = jsonObj.getString("table");

                                // 判断是否为维度表的数据
                                TableProcessDim tableProcessDim = null;
                                if ((tableProcessDim = broadcastState.get(tableName)) != null
                                        || (tableProcessDim = configMap.get(tableName)) != null) {

                                    //只需要将 data 中的数据写出， 提取data
                                    JSONObject dataJsonObj = jsonObj.getJSONObject("data");

                                    //删除不需要的字段
                                    //从tableProcessDim里面提取需要的字段
                                    List<String> needColumns = Arrays.asList(tableProcessDim.getSinkColumns().split(","));
                                    dataJsonObj.keySet().removeIf(key -> !needColumns.contains(key));

                                    //补充type ,下游在将数据写入到hbase的时候，需要基于type决定做 put 还是 delete
                                    dataJsonObj.put("type", jsonObj.getString("type"));

                                    //写出数据
                                    out.collect(Tuple2.of(dataJsonObj, tableProcessDim));

                                }
                            }

                            /**
                             * 对广播流 ( 配置流 )中的数据进行处理，
                             *
                             * 将广播流中过来的数据维护到广播状态中
                             *
                             *  r c u 的操作， 更新（添加或者覆盖）状态中的数据
                             *
                             *  d的操作， 从状态中删除对应的数据
                             */
                            @Override
                            public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
                                //获取广播状态
                                BroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                                //状态中的key
                                String key = tableProcessDim.getSourceTable();
                                //操作类型
                                String op = tableProcessDim.getOp();

                                if (Constant.CDC_OP_D.equals(op)) {
                                    //d 的操作
                                    //从状态中将当前key对应的数据清除掉
                                    broadcastState.remove(key);

                                    //从configMap中删除
                                    configMap.remove(key);
                                } else {
                                    // c r u 的操作
                                    //更新状态中的key的数据
                                    broadcastState.put(key, tableProcessDim);
                                    //更新configMap中key的数据
                                    configMap.put(key, tableProcessDim);
                                }
                            }
                        }
                );
//        filterDimDataDs.print();
        return filterDimDataDs;
    }

    private static SingleOutputStreamOperator<TableProcessDim> createOrDropTable(SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs) {
        SingleOutputStreamOperator<TableProcessDim> createOrDropTableDs = tableProcessDimDs.process(
                new ProcessFunction<TableProcessDim, TableProcessDim>() {

                    private Connection connection;

                    /**
                     * 在open方法中，获取Hbase的Connection， 保证每个并行实例中只获取一次Connection对象
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connection = HBaseUtil.getConnection();
                    }

                    /**
                     * 在close方法中，关闭Hbase的Connection
                     */
                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeConnection(connection);
                    }

                    @Override
                    public void processElement(TableProcessDim tableProcessDim, ProcessFunction<TableProcessDim, TableProcessDim>.Context ctx, Collector<TableProcessDim> out) throws Exception {
                        //根据TableProcessDim中的op决定建表还是删除表
                        //提取类型
                        String op = tableProcessDim.getOp();
                        //提取SinkTable
                        String sinkTableName = tableProcessDim.getSinkTable();
                        //提取列族
                        String sinkFamily = tableProcessDim.getSinkFamily();
                        if (Constant.CDC_OP_D.equals(op)) {
                            // op: d  -> 删表
                            HBaseUtil.dropTable(connection, Constant.HBASE_NAMESPACE, sinkTableName);
                        } else if (Constant.CDC_OP_U.equals(op)) {
                            // op: u  -> 先删表再建表
                            HBaseUtil.dropTable(connection, Constant.HBASE_NAMESPACE, sinkTableName);
                            HBaseUtil.createTable(connection, Constant.HBASE_NAMESPACE, sinkTableName, sinkFamily);
                        } else {
                            // op: r  -> 建表
                            // op: c  -> 建表
                            HBaseUtil.createTable(connection, Constant.HBASE_NAMESPACE, sinkTableName, sinkFamily);
                        }

                        //写出数据
                        out.collect(tableProcessDim);
                    }
                }
        ).setParallelism(1);
        return createOrDropTableDs;
    }

    private static SingleOutputStreamOperator<TableProcessDim> readTableProcessDim(StreamExecutionEnvironment env) {
        // 使用FlinkCDC读取 维度配置表 , 转换成 TableProcessDim 类型
        // 如何维护维度表:
        //  方式一: 通过自定义集合，维护所有的维度表   [activity_info , activity_rule , activity_sku ....]

        //  方式二: 定义到配置文件中 ，可以灵活修改配置文件， 需要解决如何让程序重新加载配置文件，让配置中的内容生效

        //  方式三: 定义到数据库表中 ，可以灵活修改表中的数据， 需要解决当数据库表中的数据发生改变后，如何让程序重新读取新的数据
        //实现方式:
        //   1. 每到达一条数据，都到数据库中读取一次最新的维度表信息

        //   2. 如果数据库的数据发生变化，需要读取， 不发生变化，不用读取

        //       数据发生变化 -> binlog ->  maxwell -> kafka  -> KafkaSource -> stream
        //       数据发生变化 -> binlog ->  FlinkCDC ->  stream

        // 方式四:  zookeeper  : 监听 + 通知

        // 注意:  在mysql中创建 gmall2024_config_240228 数据库， 在数据库中创建 table_process_dim表，
        //       在Mysql的 /etc/my.cnf文件中添加 binlog-do-db=gmall2024_config_240228
        //
        //       对Mysql的/etc/my.cnf 文件的修改会导致Maxwell出问题，需要将 maxwell在mysql数据库中的表删除掉，重新启动maxwell

        MySqlSource<String> mysqlSource =
                FlinkSourceUtil.getMySqlSource(Constant.GMALL_CONFIG_DB, Constant.TABLE_PROCESS_DIM);
        DataStreamSource<String> configDs =
                env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource")
                        .setParallelism(1);
        SingleOutputStreamOperator<TableProcessDim> tableProcessDimDs = configDs.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String value) throws Exception {
                        // r:  数据在after中
                        // c:  数据在after中
                        // u:  数据在after中
                        // d:  数据在before中

                        //将数据转换成Json格式
                        JSONObject jsonObj = JSON.parseObject(value);
                        //提取操作类型
                        String op = jsonObj.getString("op");

                        TableProcessDim tableProcessDim = null;
                        if (Constant.CDC_OP_D.equals(op)) {
                            //从before中提取数据
                            tableProcessDim = jsonObj.getObject("before", TableProcessDim.class);
                        } else {
                            // c r u
                            // 从after中提取数据
                            tableProcessDim = jsonObj.getObject("after", TableProcessDim.class);
                        }

                        //补充op
                        tableProcessDim.setOp(op);

                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        return tableProcessDimDs;
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            //将数据转换成Json格式
                            JSONObject jsonObj = JSON.parseObject(value);
                            //提取database
                            String database = jsonObj.getString("database");
                            //提取操作类型
                            String type = jsonObj.getString("type");
                            //提取data
                            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

                            if (
                                    Constant.DW_DB.equals(database)
                                            &&
                                            (
                                                    Constant.MAXWELL_TYPE_BOOTSTRAP_INSERT.equals(type)
                                                            ||
                                                            Constant.MAXWELL_TYPE_INSERT.equals(type)
                                                            ||
                                                            Constant.MAXWELL_TYPE_UPDATE.equals(type)
                                                            ||
                                                            Constant.MAXWELL_TYPE_DELETE.equals(type)
                                            )
                                            &&
                                            dataJsonObj != null && dataJsonObj.size() >= 2

                            ) {
                                out.collect(jsonObj);
                            }

                        } catch (Exception e) {
                            System.out.println("DIM 脏数据: " + value);
                            //log.warn("DIM 脏数据: "  + value );
                        }
                    }
                }
        );
        return etlStream;
    }
}